package com.heima.article.stream;

import com.alibaba.fastjson.JSON;
import com.heima.model.mess.UpdateArticleMess;
import com.heima.model.mess.pojos.ArticleVisitStreamMess;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;

@Configuration
@Slf4j
public class KafkaStreamHelloListener {

    @Bean
    public KStream<String, String> kStream(StreamsBuilder streamsBuilder) {
        KStream<String, String> stream = streamsBuilder.stream("itcast-topic-in");

        stream.map((key, value) -> {
            if (StringUtils.isEmpty(value)) {
                return new KeyValue<>(key, value);
            }

            UpdateArticleMess mess = JSON.parseObject(value, UpdateArticleMess.class);
            return new KeyValue<>(mess.getArticleId().toString(), mess.getType() + ":" + mess.getAdd());
        })
                .groupBy((key, value) -> key)
                .windowedBy(TimeWindows.of(Duration.ofSeconds(5)))
                .aggregate(
                        new Initializer<String>() {
                            @Override
                            public String apply() {
                                return "COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0";
                            }
                        },

                        new Aggregator<String, String, String>() {
                            @Override
                            public String apply(String key, String value, String initData) {
                                String[] split = value.split(":");
                                String type = split[0];
                                String add = split[1];

                                Map<String, Integer> map = string2Map(initData);
                                Integer newValue = map.get(type) + Integer.parseInt(add);
                                map.put(type, newValue);

                                return map2String(map);
                            }
                        },

                        Materialized.as("hot-atricle-stream-count-001")
                )
                .toStream()
                .map((key, value) -> {
                    return new KeyValue<>(key.key().toString(), string2JsonString(value, Long.parseLong(key.key())));
                })
                .to("itcast-topic-out");

        return stream;
    }

    /**
     * 字符串转实体类
     *
     * @param str       COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0
     * @param articleId 文章id
     * @return ArticleVisitStreamMess
     */
    private String string2JsonString(String str, Long articleId) {
        Map<String, Integer> map = string2Map(str);

        ArticleVisitStreamMess mess = new ArticleVisitStreamMess();
        mess.setArticleId(articleId);
        mess.setCollect(map.get("COLLECTION"));
        mess.setComment(map.get("COMMENT"));
        mess.setLike(map.get("LIKES"));
        mess.setView(map.get("VIEWS"));

        return JSON.toJSONString(mess);
    }

    /**
     * Map转字符串
     *
     * @param map map
     * @return COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0
     */
    private String map2String(Map<String, Integer> map) {
        Set<String> keySet = map.keySet();
        StringBuilder builder = new StringBuilder();
        for (String key : keySet) {
            builder.append(key).append(":").append(map.get(key));
            builder.append(",");
        }

        String result = builder.toString();
        return result.substring(0, result.length() - 1);
    }

    /**
     * 字符串转Map结构
     *
     * @param str COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0
     * @return Map
     */
    private Map<String, Integer> string2Map(String str) {
        Map<String, Integer> map = new HashMap<>();

        String[] initDataArray = str.split(",");
        for (String data : initDataArray) {

            String[] dataArray = data.split(":");
            String dataKey = dataArray[0];
            String dataValue = dataArray[1];

            map.put(dataKey, Integer.parseInt(dataValue));
        }

        return map;
    }
}